package com.redis.springbootredis.rabbitmq.test;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @author
 * @create 2020-11-03 20:13
 * 消息属性测试的消费者
 */
public class MessageConsumer {

    public static void main(String[] args) throws Exception {
        //创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置访问的ip地址
        connectionFactory.setHost("192.168.238.132");
        //设置虚拟主机(类似于redis的db)
        connectionFactory.setVirtualHost("ljw");
        //设置密码和账户
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //新建连接对象
        Connection connection = connectionFactory.newConnection();
        //创建连接信道(需要通信的话必须创建，类似于输送水的水管)
        Channel channel = connection.createChannel();
        //定义交换机
        /*
          String exchange 交换机名称
          BuiltinExchangeType type 交换机类型
          boolean durable 是否持久
        */
        channel.exchangeDeclare("exchangemessage", BuiltinExchangeType.TOPIC,true);
        //定义队列
        /*
         String queue  定义队列的名称
         boolean durable 队列是否持久化，即重启rabbitMq后是否保存队列
         boolean exclusive 是否为独占队列，一般为false，因为别的连接无法使用，除非自己生产自己消费可用
         boolean autoDelete  队列使用完后是否删除，一般也是false
         Map<String, Object> arguments  创建队列时附带的其他参数
        */
        Map<String, Object> arguments = new HashMap<>();
        //指定队列中所有消息的过期时间是5秒
//        arguments.put("x-message-ttl",5000);
        //指定队列存放的最大消息数(当生产者发送大于此限定的数量时时无法存入队列的)
        arguments.put("x-max-length",8);
        channel.queueDeclare("MessageQueue",true,
                false,false,arguments);
        //将队列绑定到交换机
        channel.queueBind("MessageQueue","exchangemessage","#.message.#");
        //ACK限流
        /*
          int prefetchCount 消费者每次取出消息的个数
          ACK限流模式下要将自动签收改为手动签收
        */
        channel.basicQos(2);
        //创建消费者
        /*
          String queue 队列名
          boolean autoAck
            手动确定：接收消息处理以后，手动编码回复服务器，服务器接收到回复确认后才会从队列中删除消息
            自动确定：只要消费者这边一进入消息处理的回调函数，服务器就会从队列中删除消息，如果执行过程中遇到异常，那么消息会丢失
          Consumer callback 处理消息的回调函数，接收消息以后执行处理的代码
        */
        channel.basicConsume("MessageQueue", false, new DefaultConsumer(channel) {

            /*
             String consumerTag 消费标记，就是一个编号
             Envelope envelope 信封，手动恢复时需要使用
             AMQP.BasicProperties properties  消费时的附加属性
             byte[] body 消费体
            */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                 //输出信息查看
                System.out.println("consumerTag---"+consumerTag);
                System.out.println("body---"+new String(body));
                System.out.println("properties---"+properties.getHeaders());
                System.out.println("消费成功");
                //ACK重回测试
//                if(new String(body).contains("0")){
//                    //不重回队列
//                   channel.basicReject(envelope.getDeliveryTag(),false);
//                   System.out.println("处理失败");
//                    //重回队列
//                   channel.basicNack(envelope.getDeliveryTag(),false,true);
//                   System.out.println("处理失败");
//                }else{
//                   //服务的回复，不确认的话消费者无法获得下一次的消费
//                   channel.basicAck(envelope.getDeliveryTag(),false);
//                }
                //服务器的应答，不确认的话消费者无法获得下一次的消费
                channel.basicAck(envelope.getDeliveryTag(),false);
            }
        });
    }
}
